package com.example.status;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: OperatorListStateDemo
 * Package: com.example.status
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-27
 * Time: 18:27
 */

//算子状态
public class OperatorListStateDemo {
    //需求 在map算子中 计算需求的个数
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        SingleOutputStreamOperator<Long> data = env
                .socketTextStream("hadoop102", 7777)
                .map(new MyMap());
        data.print();
        env.execute();
    }

    //在map算子中 计算需求的个数
    //Map算子本身是无状态的算子
    //现在要自定义实现状态 实现CheckpointedFunction接口
    public static class MyMap implements MapFunction<String, Long>, CheckpointedFunction {

        //要实现 奇数 定义一个遍历来计数
        private long count = 0L;
        //要实现状态化 就是将这个变量 存储到状态中去

        //2.1.1先要创建一个状态
        private ListState<Long> state;

        @Override
        public Long map(String value) throws Exception {
            //直接++
            return ++count;
        }

        /**
         * 将状态做快照 本地变量持久化
         * 定义 将 本地变量 拷贝到 算子状态中
         * @param context
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("调用 snapshotState");
            //2.1清空状态
            state.clear();
            //将count值添加到状态中
            state.add(count);

        }
        /**
         * 初始化本地变量 从状态中 把数据添加到本地 变量 每个子任务调用一次
         * 此方法用在程序恢复的时候 比如程序挂了 变量是在内存当中数据丢失了
         * 那么从状态中恢复之前的数据
         * 开启checkpoint时候才会调用
         * @param context
         * @throws Exception
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("调用initializeState");
            //3.1获取算子 状态
            //从上下文中 获取算子状态
            context.getOperatorStateStore()
                    .getListState(new ListStateDescriptor<Long>("state", Types.LONG));

            //3.2从算子状态中 把数据 拷贝到 本地变量
            //先做一个判断 判断上下文是否成功了
            if (context.isRestored()) {
                //从状态中把数据拷贝过来
                for (Long c : state.get()) {
                    count += c;
                }
            }
        }
    }
}
